package com.eshipenet.shipnet.equipmentextbuilder

import java.lang

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{MapStateDescriptor, ValueStateDescriptor}
import org.apache.flink.runtime.state.KeyedStateFunction
import org.apache.flink.streaming.api.datastream.CoGroupedStreams.TaggedUnion
import org.apache.flink.streaming.api.functions.co.{CoProcessFunction, KeyedBroadcastProcessFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{
  EventTimeSessionWindows,
  GlobalWindows,
  ProcessingTimeSessionWindows,
  SessionWindowTimeGapExtractor
}
import org.apache.flink.streaming.api.windowing.evictors.Evictor
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow
import org.apache.flink.streaming.runtime.operators.windowing.TimestampedValue
import org.apache.flink.util.Collector

import scala.collection.JavaConverters._

object Processor {

  def generateEquipmentWithModelIdStream(
      equipmentActionStream: DataStream[EquipmentAction],
      joinStream: DataStream[EquipmentModelLinkAction]): DataStream[EquipmentWithModelIdAction] = {
    equipmentActionStream
      .connect(joinStream)
      .keyBy(f => f.id, f => f.eid)
      .process(new EquipmentWithModelIdActionGenerator)
  }

  def generateEquipmentWithModelActionStream(
      equipmentWithModelIdActionStream: DataStream[EquipmentWithModelIdAction],
      equipmentModelActionStream: DataStream[EquipmentModelAction]): DataStream[EquipmentWithModelAction] = {
    equipmentWithModelIdActionStream
      .keyBy(_.id)
      .join(equipmentModelActionStream.keyBy(_.id))
      .where(_.modelId)
      .equalTo(_.id)
      .window(GlobalWindows.create())
      .trigger(new Trigger[TaggedUnion[EquipmentWithModelIdAction, EquipmentModelAction], GlobalWindow] {
        override def onElement(element: TaggedUnion[EquipmentWithModelIdAction, EquipmentModelAction],
                               timestamp: Long,
                               window: GlobalWindow,
                               ctx: Trigger.TriggerContext): TriggerResult = {
          TriggerResult.FIRE
        }

        override def onEventTime(time: Long, window: GlobalWindow, ctx: Trigger.TriggerContext): TriggerResult =
          TriggerResult.CONTINUE

        override def onProcessingTime(time: Long, window: GlobalWindow, ctx: Trigger.TriggerContext): TriggerResult =
          TriggerResult.CONTINUE

        override def clear(window: GlobalWindow, ctx: Trigger.TriggerContext): Unit = Unit
      })
      .evictor(new Evictor[TaggedUnion[EquipmentWithModelIdAction, EquipmentModelAction], GlobalWindow] {
        override def evictAfter(
            elements: lang.Iterable[TimestampedValue[TaggedUnion[EquipmentWithModelIdAction, EquipmentModelAction]]],
            size: Int,
            window: GlobalWindow,
            evictorContext: Evictor.EvictorContext): Unit = Unit
        override def evictBefore(
            elements: lang.Iterable[TimestampedValue[TaggedUnion[EquipmentWithModelIdAction, EquipmentModelAction]]],
            size: Int,
            window: GlobalWindow,
            evictorContext: Evictor.EvictorContext): Unit = {
          val lst          = elements.iterator().asScala.toList
          val lastIdAct    = lst.lastIndexWhere(_.getValue.isOne)
          val lastModelAct = lst.lastIndexWhere(_.getValue.isTwo)
          val it           = elements.iterator()
          var index        = 0
          for (_ <- it.asScala) {
            if (index != lastIdAct && index != lastModelAct) {
              it.remove()
            }
            index = index + 1
          }
        }
      })
      .apply((eid, m, out: Collector[EquipmentWithModelAction]) => {
        (eid, m) match {
          case (EquipmentWithModelIdInsert(pair), EquipmentModelInsert(_, model)) =>
            val e = EquipmentWithModel(pair.equipment, model, pair.link)
            out.collect(EquipmentWithModelInsert(e))
          case (EquipmentWithModelIdInsert(pair), EquipmentModelUpdate(_, _, model)) =>
            val e = EquipmentWithModel(pair.equipment, model, pair.link)
            out.collect(EquipmentWithModelInsert(e))
          case (EquipmentWithModelIdInsert(pair), EquipmentModelDelete(_, model)) =>
            val e = EquipmentWithModel(pair.equipment, model, pair.link)
            out.collect(EquipmentWithModelDelete(e))
          case (EquipmentWithModelIdDelete(pair), EquipmentModelInsert(_, model)) =>
            val e = EquipmentWithModel(pair.equipment, model, pair.link)
            out.collect(EquipmentWithModelDelete(e))
          case (EquipmentWithModelIdDelete(pair), EquipmentModelUpdate(_, _, model)) =>
            val e = EquipmentWithModel(pair.equipment, model, pair.link)
            out.collect(EquipmentWithModelDelete(e))
          case (EquipmentWithModelIdDelete(pair), EquipmentModelDelete(_, model)) =>
            val e = EquipmentWithModel(pair.equipment, model, pair.link)
            out.collect(EquipmentWithModelDelete(e))
        }
      })
  }

//  def generateEquipmentWithModelActionStream(
//      equipmentWithModelIdActionStream: DataStream[EquipmentWithModelIdAction],
//      equipmentModelActionStream: DataStream[EquipmentModelAction]): DataStream[EquipmentWithModelAction] = {
//    val desc =
//      new MapStateDescriptor[String, EquipmentModel]("EquipmentModel", classOf[String], classOf[EquipmentModel])
//    equipmentWithModelIdActionStream
//      .keyBy(f => s"${f.id}.${f.modelId})")
//      .connect(equipmentModelActionStream.broadcast(desc))
//      .process(
//        new KeyedBroadcastProcessFunction[String,
//                                          EquipmentWithModelIdAction,
//                                          EquipmentModelAction,
//                                          EquipmentWithModelAction] {
//          private lazy val equipmentWithModelIdDesc =
//            new ValueStateDescriptor[Option[(Equipment, String)]]("EquipmentWithModelId",
//                                                                  classOf[Option[(Equipment, String)]])
//
//          override def processBroadcastElement(value: EquipmentModelAction,
//                                               ctx: KeyedBroadcastProcessFunction[String,
//                                                                                  EquipmentWithModelIdAction,
//                                                                                  EquipmentModelAction,
//                                                                                  EquipmentWithModelAction]#Context,
//                                               out: Collector[EquipmentWithModelAction]): Unit = {
//            val state = ctx.getBroadcastState(desc)
//            value match {
//              case EquipmentModelInsert(_, after) =>
//                state.put(after.id, after)
//                ctx.applyToKeyedState(
//                  equipmentWithModelIdDesc,
//                  new KeyedStateFunction[String, Option[(Equipment, String)]] {
//                    override def process(key: String, state: Option[(Equipment, String)]): Unit = {
//                      println("apply")
////                      out.collect(EquipmentWithModelInsert(EquipmentWithModel(after)))
//                    }
//                  }
//                )
//              case EquipmentModelUpdate(_, _, after) =>
//                state.put(after.id, after)
//              case EquipmentModelDelete(_, before) =>
//                state.remove(before.id)
//            }
//          }
//
//          override def processElement(value: EquipmentWithModelIdAction,
//                                      ctx: KeyedBroadcastProcessFunction[String,
//                                                                         EquipmentWithModelIdAction,
//                                                                         EquipmentModelAction,
//                                                                         EquipmentWithModelAction]#ReadOnlyContext,
//                                      out: Collector[EquipmentWithModelAction]): Unit = {
//            val state = ctx.getBroadcastState(desc)
//            value match {
//              case EquipmentWithModelIdInsert(after) =>
//                Option(state.get(after._2)).foreach { v =>
//                  val e      = EquipmentWithModel(after._1, v)
//                  val insert = EquipmentWithModelInsert(e)
//                  out.collect(insert)
//                }
//              case EquipmentWithModelIdDelete(before) =>
//                Option(state.get(before._2)).foreach { v =>
//                  val e      = EquipmentWithModel(before._1, v)
//                  val delete = EquipmentWithModelDelete(e)
//                  out.collect(delete)
//                }
//            }
//          }
//
//        })
//  }

//  def generateEquipmentWithModelActionStream(
//      equipmentWithModelIdActionStream: DataStream[EquipmentWithModelIdAction],
//      equipmentModelActionStream: DataStream[EquipmentModelAction]): DataStream[EquipmentWithModelAction] = {
//    equipmentWithModelIdActionStream.map()
//  }

  def mergeToEquipmentExtAction(stream: DataStream[EquipmentWithModelAction]): DataStream[EquipmentExtAction] =
    stream
      .keyBy(_.id)
      .mapWithState((action, state: Option[EquipmentExt]) => {
        action match {
          case EquipmentWithModelInsert(after) =>
            val s   = state.getOrElse(EquipmentExt(after.equipment, Map()))
            val ns  = s.copy(models = s.models.updated(after.model.id, after.model))
            val ext = EquipmentExtInsert(ns)
            (Some(ext), Some(ns))
          case EquipmentWithModelDelete(before) =>
            state match {
              case Some(s) =>
                val ns = s.copy(models = s.models - before.model.id)
                if (ns.models.isEmpty) {
                  val ext = EquipmentExtDelete(ns)
                  (Some(ext), None)
                } else {
                  val update = EquipmentExtUpdate(s, ns)
                  (Some(update), Some(ns))
                }
              case _ =>
                (None: Option[EquipmentExtAction], None)
            }
        }
      })
      .filter(_.isDefined)
      .map(_.get)
      .keyBy(_.id)
      .window(ProcessingTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor[EquipmentExtAction] {
        override def extract(element: EquipmentExtAction): Long = element match {
          case _: EquipmentExtInsert => 5000
          case _: EquipmentExtUpdate => 5000
          case _: EquipmentExtDelete => 1
        }
      }))
      .aggregate(new AggregateFunction[EquipmentExtAction, Option[EquipmentExtAction], Option[EquipmentExtAction]] {
        override def createAccumulator(): Option[EquipmentExtAction] = None

        override def add(value: EquipmentExtAction,
                         accumulator: Option[EquipmentExtAction]): Option[EquipmentExtAction] = {
          Some(value)
        }

        override def merge(a: Option[EquipmentExtAction], b: Option[EquipmentExtAction]): Option[EquipmentExtAction] = {
          (a, b) match {
            case (Some(_), Some(y)) => Some(y)
            case (Some(x), None)    => Some(x)
            case (None, Some(y))    => Some(y)
            case _                  => None
          }
        }

        override def getResult(accumulator: Option[EquipmentExtAction]): Option[EquipmentExtAction] = accumulator
      })
      .filter(_.isDefined)
      .map(_.get)
//      .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))
//      .aggregate(new AggregateFunction[EquipmentWithModelAction, List[EquipmentWithModelAction], EquipmentExtAction] {
//        override def createAccumulator(): List[EquipmentWithModelAction] = List()
//        override def add(value: EquipmentWithModelAction,
//                         accumulator: List[EquipmentWithModelAction]): List[EquipmentWithModelAction] =
//          accumulator :+ value
//
//        override def merge(a: List[EquipmentWithModelAction],
//                           b: List[EquipmentWithModelAction]): List[EquipmentWithModelAction] =
//          a ++ b
//
//        override def getResult(accumulator: List[EquipmentWithModelAction]): EquipmentExtAction = {
//          val s = accumulator
//            .groupBy(v => (v.id, v.modelId))
//            .mapValues(_.last)
//            .values
//            .groupBy {
//              case _: EquipmentWithModelInsert => "insert"
//              case _: EquipmentWithModelDelete => "delete"
//            }
//            .mapValues(_.toList)
//          s.get("insert") match {
//            case Some(lst) => buildEquipmentExtInsert(lst.map(_.asInstanceOf[EquipmentWithModelInsert]))
//            case _         => EquipmentExtDelete(accumulator.head.id)
//          }
//        }
//      })

//  def buildEquipmentExtInsert(actions: List[EquipmentWithModelInsert]): EquipmentExtInsert = {
//    val v = actions
//      .foldLeft(None: Option[(Equipment, Set[EquipmentModel])]) { (sum, act) =>
//        val cs = sum.getOrElse((act.after.equipment, Set[EquipmentModel]()))
//        Some((cs._1, cs._2 + act.after.model))
//      }
//      .get
//    EquipmentExtInsert(v._1.id, v._1, v._2)
//  }

}
